/**** 1. 核心部分 ****/
export class Observable {
  constructor(subscribe_fn) {
    this.subscribe_fn = subscribe_fn; // 订阅函数（给每个observer执行）
  }
  subscribe(raw_observer) {
    let observer = {
      next: (value) => {},
      error: (error) => {},
      complete: () => {},
    }
    if(typeof raw_observer === 'function') {
      observer.next = raw_observer
    } else if (raw_observer instanceof Subject) {
      observer = raw_observer
    } else if(typeof raw_observer === 'object') {
      // fix：raw_observer有可能是subject实例，该实例对应方法是定义在原型上的，解构不到(ps: 上面兼容了，fix：this指向不一致问题)
      // observer = {...observer, ...raw_observer}
      Object.keys(observer).forEach(key => {
        if(raw_observer[key]) {
          observer[key] = raw_observer[key]
        }
      })
    }

    let subscription = {
      unsubscribe: this.subscribe_fn(observer) || (() => {}),
    }
    return subscription;
  }
  // version 1.0.0
  pipe(...operators) {
    let root_observable = this
    return new Observable(subscriber => {
      let opindex = 0, curr_observable = root_observable
      // 从左到右依次执行operators，构建一条observable管道链（后一个内部订阅前一个）
      while(opindex < operators.length) {
        curr_observable = operators[opindex](curr_observable)
        opindex++
      }
      // operators的执行顺序：从左到右
      // 但是订阅顺序（subscribe的执行顺序）：从右到左（一旦开始订阅，observable对应的订阅函数subscribe_fn会依次执行）
      return curr_observable.subscribe(subscriber).unsubscribe // 因为本身不需要定义unsubscribe，所以直接返回上一个observable的unsubscribe
    })
  }
  // version 2.0.0（去掉多余一层Observable）
  pipe(...operators) {
    let opindex = 0, curr_observable = this
    while(opindex < operators.length) {
      curr_observable = operators[opindex](curr_observable)
      opindex++
    }
    return curr_observable
  }
}

// 管道操作符1
export function map(fn) {
  return (observable) => {
    // subscriber参数 对应订阅我的observer提供的系列预处理函数
    return new Observable(subscriber => {
      return observable.subscribe({ // observable.subscribe 对应创建一个observer，去订阅其它的observable
        next: (value) => subscriber.next(fn(value)), // 其它observable返回值，接着我处理过后，再返回给订阅我的observer
        // 错误处理也要维护好管道链式关系
        error: (error) => subscriber.error(error),
        complete: () => subscriber.complete(),
      }).unsubscribe
    })
  }
}

// 管道操作符2
export function delay(delayInMillis) {
  return (observable) => {
    return new Observable(subscriber => {
      const allTimerIDs = new Set();
      let hasCompleted = false;
      const subscription = observable.subscribe({
        next(value) {
          const timerID = setTimeout(() => {
            subscriber.next(value);
            allTimerIDs.delete(timerID);
            if (hasCompleted && allTimerIDs.size === 0) {
              subscriber.complete();
            }
          }, delayInMillis);
 
          allTimerIDs.add(timerID);
        },
        error(err) {
          subscriber.error(err);
        },
        complete() {
          hasCompleted = true;
          if (allTimerIDs.size === 0) {
            subscriber.complete();
          }
        },
      })
      return () => {
        subscription.unsubscribe();
        for (const timerID of allTimerIDs) {
          clearTimeout(timerID);
        }
      };
    })
  }
}

// creation operator 1
export function of(...values) {
  return new Observable(subscriber => {
    for(let i = 0; i < values.length; i++) {
      subscriber.next(values[i])
    }
    subscriber.complete()
  })
}


/**** 2. 扩展部分 ****/
export class Observer {
  constructor(raw_observer) {
    if(typeof raw_observer === 'function') {
      this.next = raw_observer
    } else if(typeof raw_observer === 'object') {
      Object.keys(raw_observer).forEach(key => {
        this[key] = raw_observer[key]
      })
    }
  }
  next(value) {}
  error(error) {}
  complete() {}
}

export class Subject extends Observer {
  constructor() {
    super()
    this.observers = []
  }

  // 收集observer
  subscribe(observer) { // 特殊点
    this.observers.push(observer)
    return {
      unsubscribe: () => this.observers.splice(this.observers.indexOf(observer), 1)
    }
  }

  next(value) {
    this.observers.forEach(observer => observer.next && observer.next(value))
  }
  error(error) {
    this.observers.forEach(observer => observer.error && observer.error(error))
  }
  complete() {
    this.observers.forEach(observer => observer.complete && observer.complete())
  }
}

// export class BehaviorSubject extends Subject;
// export class ReplaySubject extends Subject;
// export class AsyncSubject extends Subject;

// creation operator 2
export function interval(delay) {
  return new Observable(subscriber => {
    let count = 0
    const timerID = setInterval(() => {
      subscriber.next(count++)
    }, delay)
    return () => clearInterval(timerID)
  })
}

export class ConnectableObservable extends Observable {
  constructor(source_observable, subject) {
    super(subscriber => {
      return subject.subscribe(subscriber).unsubscribe
    })
    this.source_observable = source_observable
    this.subject = subject
  }
  connect() {
    return this.source_observable.subscribe(this.subject) // 因为subject实例本身也是observer，所以可以订阅observable
  }
}

// 管道操作符3 --- 多播操作符
export function multicast(subject) {
  return (observable) => {
    return new ConnectableObservable(observable, subject)
  }
}

// 管道操作符4 --- 引用计数操作符(实现自动控制ConnectableObservable的连接和断开)
export function refCount() {
  return (connectable_observable) => {
    if(connectable_observable instanceof ConnectableObservable) {
      let source_subscription = null // ps:注意作用域
      return new Observable(subscriber => {
        const subscription = connectable_observable.subscribe(subscriber)
        if(connectable_observable.subject.observers.length === 1) {
          source_subscription = connectable_observable.connect() // 返回unsubscribe
        }
        return () => { // unsubscribe增强
          subscription.unsubscribe()
          if(connectable_observable.subject.observers.length === 0) {
            source_subscription.unsubscribe()
          }
        }
      })
    }
    return connectable_observable
  }
}



/**** 3. 测试用例 ****/

// ----- 1. 基础用法 -----
// of(1, 2, 3).pipe(map(x => x * 2), delay(1000)).subscribe({
//   next: (value) => console.log(value),
//   error: (error) => console.error(error),
//   complete: () => console.log('complete'),
// })



// ----- 2. Subject实例实现多播 -----
// const subject = new Subject();
// subject.subscribe({
//   next: (value) => console.log(value),
// })
// subject.subscribe({
//   next: (value) => console.log(value),
// })
// subject.subscribe({
//   next: (value) => console.log(value),
// })
// subject.next(1)



// ----- 3. 多播官方测试用例（手动控制）-----
// const source = interval(500);
// const subject = new Subject();
// const multicasted = source.pipe(multicast(subject));
// let subscription1, subscription2, subscriptionConnect;
 
// subscription1 = multicasted.subscribe({
//   next: (v) => console.log(`observerA: ${v}`),
// });
// // We should call `connect()` here, because the first
// // subscriber to `multicasted` is interested in consuming values
// subscriptionConnect = multicasted.connect();
 
// setTimeout(() => {
//   subscription2 = multicasted.subscribe({
//     next: (v) => console.log(`observerB: ${v}`),
//   });
// }, 600);
 
// setTimeout(() => {
//   subscription1.unsubscribe();
// }, 1200);
 
// // We should unsubscribe the shared Observable execution here,
// // because `multicasted` would have no more subscribers after this
// setTimeout(() => {
//   subscription2.unsubscribe();
//   subscriptionConnect.unsubscribe(); // for the shared Observable execution
// }, 2000);



// ----- 4. 多播官方测试用例（依赖refCount自动控制）-----
const source = interval(500);
const subject = new Subject();
const refCounted = source.pipe(multicast(subject), refCount());
let subscription1, subscription2;
 
// This calls `connect()`, because
// it is the first subscriber to `refCounted`
console.log('observerA subscribed');
subscription1 = refCounted.subscribe({
  next: (v) => console.log(`observerA: ${v}`),
});
 
setTimeout(() => {
  console.log('observerB subscribed');
  subscription2 = refCounted.subscribe({
    next: (v) => console.log(`observerB: ${v}`),
  });
}, 600);
 
setTimeout(() => {
  console.log('observerA unsubscribed');
  subscription1.unsubscribe();
}, 1200);
 
// This is when the shared Observable execution will stop, because
// `refCounted` would have no more subscribers after this
setTimeout(() => {
  console.log('observerB unsubscribed');
  subscription2.unsubscribe();
}, 2000);
